package com.zheng.flink.study.dataset.transformation;

import org.apache.flink.api.common.functions.FlatJoinFunction;
import org.apache.flink.api.common.functions.JoinFunction;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.util.Collector;

import java.util.ArrayList;
import java.util.List;

/**
 * Created by zhengbo on 2019/12/13.
 */
public class JoinTransformation {

    public static void main(String[] args) throws Exception {

        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

        List<Tuple2<Integer, String>> info1 = new ArrayList<>();
        info1.add(new Tuple2<>(1, "zhangyi"));
        info1.add(new Tuple2<>(2, "hyq"));
        info1.add(new Tuple2<>(3, "dzz"));
        info1.add(new Tuple2<>(4, "zhangsananan"));

        List<Tuple2<Integer, String>> info2 = new ArrayList<>();
        info2.add(new Tuple2<>(1, "杭州"));
        info2.add(new Tuple2<>(2, "上海"));
        info2.add(new Tuple2<>(3, "杭州"));
        info2.add(new Tuple2<>(5, "杭州"));


        DataSource<Tuple2<Integer, String>> data1 = env.fromCollection(info1);

        DataSource<Tuple2<Integer, String>> data2 = env.fromCollection(info2);

        //where data1.tuple.f0=data.tuple.f0 即两边id一致做join
        data1.join(data2).where(new KeySelector<Tuple2<Integer, String>, Integer>() {
            @Override
            public Integer getKey(Tuple2<Integer, String> value) throws Exception {
                return value.f0;
            }
        }).equalTo(new KeySelector<Tuple2<Integer, String>, Integer>() {
            @Override
            public Integer getKey(Tuple2<Integer, String> value) throws Exception {
                return value.f0;
            }
        }).with(new JoinFunction<Tuple2<Integer, String>, Tuple2<Integer, String>, Tuple3<Integer, String, String>>() {
            @Override
            public Tuple3<Integer, String, String> join(Tuple2<Integer, String> first, Tuple2<Integer, String> second) throws Exception {
                return new Tuple3<>(first.f0, first.f1, second.f1);
            }
        }).print();


        //FlatJoinFunction做join
        data1.join(data2).where(new KeySelector<Tuple2<Integer, String>, Integer>() {
            @Override
            public Integer getKey(Tuple2<Integer, String> value) throws Exception {
                return value.f0;
            }
        }).equalTo(new KeySelector<Tuple2<Integer, String>, Integer>() {
            @Override
            public Integer getKey(Tuple2<Integer, String> value) throws Exception {
                return value.f0;
            }
        }).with(new FlatJoinFunction<Tuple2<Integer, String>, Tuple2<Integer, String>, Tuple4<Integer, String, String, Integer>>() {

            @Override
            public void join(Tuple2<Integer, String> first, Tuple2<Integer, String> second,
                             Collector<Tuple4<Integer, String, String, Integer>> out) throws Exception {
                out.collect(new Tuple4<>(first.f0, first.f1, second.f1, 1));
            }
            //按城市分组 求和(在当前城市总人数)
        }).groupBy(2).sum(3).print();

    }
}
